﻿using Confluent.Kafka;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace ProducerTest1
{
    public class BathPush : IPush
    {
        public void Invoke(ProducerConfig config)
        {
            using (var producer = new ProducerBuilder<string, string>(config).Build())
            {
                try
                {
                    //不要看着这里使用的是同步方法，其实这种方式对于kafka的连接来说叫做“流式推送”
                    for (int i = 0; i < 100; i++)
                    {
                        producer.Produce("CustomerCountry", new Message<string, string> { Key = "key_" + i, Value = "value_" + i });
                    }
                    producer.Flush();
                }
                catch (ProduceException<string, string> e)
                {
                    Console.WriteLine($"failed to deliver message:{e.Message} [{e.Error.Code}]");
                }
            }
        }

        public Task InvokeAsync(ProducerConfig config)
        {
            throw new NotImplementedException();
        }
    }
}
